package spark.core.scala

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

/**
 * Created by Administrator on 2018/1/31/031.
 */
object SparkCoreWC {

  /**
   * 这是spark原始的WORDCOUNT
   *
   */
  def main(args: Array[String]) {
    //1.spark上下文
    val conf = new SparkConf()
      .setMaster("local[*]")
      .setAppName("SparkCoreWC")
    val sc = new SparkContext(conf)
    //2.读取数据块，生成RDD（集合、数据集）
    val url = ("hdfs://bigdata00:8020/user/hadoop/wordcount.txt")
    val saveUrl = s"result/sparkcore_${System.currentTimeMillis()}"
    val rdd :RDD[String] = sc.textFile(url)
    //3.处理数据（第一种方式）
    /**
     * （第一种方式）使用集合的API 来做数据的WC处理
     */
//    val result = rdd.flatMap(_.split(","))
//      .map(word => (word,1))
//      .groupBy(word => word)
//      .map(t => (t._1._1,t._2.toList.size))

    /**
     * （第二种方式）使用集合的API 来做数据的WC处理
     */
      val result = rdd.flatMap(_.split(","))
      .map(word => (word,1))
      .reduceByKey(_+_)



     //4.处理完数据之后进行数据的打印，输出到控制台
      result.foreachPartition(itr =>{
      itr.foreach(println)
    })

    result.saveAsTextFile(saveUrl)
    //5.为了查看4040页面，做一个线程等待

    Thread.sleep(10000l)


  }

}
